package com.liu.kafka.config;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ConsumerAwareListenerErrorHandler;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.ListenerExecutionFailedException;
import org.springframework.kafka.support.ProducerListener;
import org.springframework.messaging.Message;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;

@Slf4j
@Configuration
public class KafkaConfiguration {
    private final static Logger logger = LoggerFactory.getLogger(KafkaConfiguration.class);

    @Autowired
    KafkaTemplate kafkaTemplate;

    @Resource
    private ConsumerFactory<String, String> consumerFactory;

    private final String ERROR_TOPIC="error_topic";

    /**
     * 监听器工厂 批量消费
     *
     * @return
     */
    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> batchFactory() {
        //获取默认配置
        Map<String, Object> properties = consumerFactory.getConfigurationProperties();
        HashMap<String, Object> props = new HashMap<>(properties);
        //每次拉取数量
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 10);
        //自动提交的时间间隔
//        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,100);
        //会话超时时间  超时触发再平衡
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10000);
        //消费超时时间
        props.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);
//        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);

        //构建消费端
        ConcurrentKafkaListenerContainerFactory<String, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(props));

        //设置为批量消费，每个批次数量在Kafka配置参数中设置ConsumerConfig.MAX_POLL_RECORDS_CONFIG
        factory.setBatchListener(true);
        factory.setConcurrency(1);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);//设置提交偏移量的方
        return factory;
    }


    /**
     * 生产者发送消息配置监听
     */
    @PostConstruct
    private void listener() {
        kafkaTemplate.setProducerListener(new ProducerListener<String, Object>() {
            /**
             * 发送成功回调
             * @param producerRecord
             * @param recordMetadata
             */
            @Override
            public void onSuccess(ProducerRecord<String, Object> producerRecord, RecordMetadata recordMetadata) {
                logger.info("send ok,message= {}", producerRecord.value());
            }

            /**
             * 发送失败监听
             * @param producerRecord
             * @param recordMetadata
             * @param exception
             */
            @Override
            public void onError(ProducerRecord<String, Object> producerRecord, RecordMetadata recordMetadata, Exception exception) {
                logger.error("send error!message= {}", producerRecord.value());
            }
        });
    }


    /**
     * 消费者自定义异常处理器
     */
//    @Bean
    public ConsumerAwareListenerErrorHandler listenerErrorHandler() {
        return new ConsumerAwareListenerErrorHandler() {

            @Override
            public Object handleError(Message<?> message, ListenerExecutionFailedException exception, Consumer<?, ?> consumer) {

                log.info("--------> message:{}", message.toString());
                log.info("Payload:{}", message.getPayload());


                log.info("exception :{}", exception.getMessage());
                exception.printStackTrace();
                System.out.println("--- 消费发生异常 ---");
                return null;
            }
        };
    }
}
